/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package hadoop

import scala.collection.mutable.{Set => MSet}
import org.apache.hadoop.conf.Configuration
import util.Base64
import util.ObjectSerializer._
import com.google.protobuf.DescriptorProtos.FileDescriptorSet
import generator.TableValue
import generator.Helpers

object SCConfiguration {
  val TABLES           = "sc.tables"
  val STATIC_VARIABLES = "sc.static_variables"
  val PROTOFILES       = "sc.pbfileset"

  private def p2o[T](p: T): Option[T] = if (p != null) Some(p) else None

  def apply(t: List[TableValue], s: Helpers.StaticVariables, p: FileDescriptorSet) = new SCConfiguration(t, s, p)
  def apply(conf: Configuration) = {
    val t = p2o(conf.get(TABLES)).map(x => deserialize(Base64.decode(x), loaders.toList)).getOrElse(Map())
    val s = p2o(conf.get(STATIC_VARIABLES)).map(x => deserialize(Base64.decode(x), loaders.toList)).getOrElse(null)
    val p = FileDescriptorSet.parseFrom(Base64.decode(p2o(conf.get(PROTOFILES)).getOrElse("")))

    new SCConfiguration(t.asInstanceOf[List[TableValue]], s.asInstanceOf[Helpers.StaticVariables], p)
  }

  val loaders = MSet[ClassLoader]();
}

/**
 * Container of data passed to Mappers and Reducers.
 *
 * @param tables           Tables information.
 * @param StaticVariables  The values of static variables.
 * @param protoFiles       Descriptors of ProtocolBuffers.
 */
class SCConfiguration(
  val tables: List[TableValue],
  val staticVariables: Helpers.StaticVariables,
  val protoFiles: FileDescriptorSet) extends NotNull {

  import SCConfiguration._

  tables.foreach { x => loaders.add(x.getClass.getClassLoader) }
  loaders.add(staticVariables.getClass.getClassLoader)

  /**
   * Sets value to Hadoop configuration.
   */
  def setTo(conf: Configuration): Unit = {
    val t = serialize(tables)
    val s = serialize(staticVariables)
    val f = protoFiles.toByteArray

    conf.set(TABLES,           Base64.encode(t))
    conf.set(STATIC_VARIABLES, Base64.encode(s))
    conf.set(PROTOFILES,       Base64.encode(f))

  }
}
